// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.utils.hbase;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.*;
import net.opentsdb.query.QueryUtil;
import net.opentsdb.uid.UniqueId;
import org.hbase.async.HBaseClient;
import org.hbase.async.KeyOnlyFilter;
import org.hbase.async.KeyValue;
import org.hbase.async.Scanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 数据表操作集合
 *
 */
public class HbaseTsdbUtils extends HbaseAbsUtils{
    private static final Logger LOG = LoggerFactory.getLogger(HbaseTsdbUtils.class);

    public HbaseTsdbUtils(final HBaseClient client, final byte[] table){
        super(client, table);
    }


    /**
     * 查询待删除的时间线
     *
     * @param tsuids
     * @return
     */
    public Set<String> getDelTsuids(Set<String> tsuids, TSDB tsdb, long start, long end){

        if(tsuids == null || tsuids.isEmpty())
            return Collections.EMPTY_SET;

        Set<String> result = Sets.newHashSet();

        TSSubQuery subQuery = new TSSubQuery();
        subQuery.setAggregator("none");
        subQuery.setTsuids(tsuids.stream().collect(Collectors.toList()));
        TSQuery query = new TSQuery();
        query.setQueries(Lists.newArrayList(subQuery));

        query.setStart(String.valueOf(start));
        query.setEnd(String.valueOf(end));
        query.validateAndSetQuery();

        query.buildQueriesAsync(tsdb).addCallback(
                iQueries -> {
                    final ArrayList<Deferred<DataPoints[]>> deferreds = new ArrayList<Deferred<DataPoints[]>>(iQueries.length);

                    for(IQuery iquery: iQueries){
                        deferreds.add(iquery.runAsync());
                    }

                    Deferred.group(deferreds).addCallback(dpsArr -> {
                       for(DataPoints[] dpsList : dpsArr){
                           for(DataPoints dps: dpsList){
                               result.addAll(dps.getTSUIDs());
                           }
                       }
                        return null;
                    });
                    return null;
                }
        );

        if(result.isEmpty()) return tsuids;

        return tsuids.stream()
                .filter(t -> !result.contains(t))
                .collect(Collectors.toSet());
    }

    /**
     * 根据时间线判断是否有数据存在
     *
     * @param tsuid
     * @return
     */
    public boolean existsDataByTsuid(String tsuid){

        boolean flag = false;
        Scanner scanner = client.newScanner(table);

        // 两者会冲突，都是filter
        scanner.setKeyRegexp(QueryUtil.getRowKeyTSUIDRegex(
                Lists.newArrayList(tsuid)), CHARSET);
//        scanner.setFilter(new KeyOnlyFilter());

        Set<byte[]> result = Sets.newHashSet();

        // TODO 需要优化，last接口不合适
        ArrayList<ArrayList<KeyValue>> rows;
        try{
            while ((rows = scanner.nextRows().joinUninterruptibly()) != null){
                for (final ArrayList<KeyValue> row : rows) {
                    if(row != null && row.size()>0)
                        result.add(row.get(0).key());
                }
            }
        }catch (Exception e){
            LOG.error("query data by tsuid error",e);
        }finally {
            if(scanner != null)
                scanner.close();
            return flag;
        }
    }

}
